package com.bst.etl.batch;

import com.bst.common.constant.JobConstant;
import com.bst.common.exception.base.BaseException;
import com.bst.common.exception.base.BaseRuntimeException;
import com.bst.common.utils.DateUtil;
import com.bst.common.utils.spring.SpringUtils;
import com.bst.etl.domain.EtlTaskExecItem;
import com.bst.etl.service.IEtlTaskExecItemService;
import org.springframework.batch.core.*;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class BatchExecService {

    @Resource(name = "jobAsyncLauncher")
    JobLauncher jobLauncher;
    @Autowired
    IEtlTaskExecItemService execItemService;

    public void runJob(JobParameters parameters, ConcurrentHashMap<Long, JobExecution> resultMap) {
        String jobType = parameters.getString(JobConstant.JOB_TYPE);
        Job job = null;
        try {
            if (JobConstant.JOB_TYPE_BASE.equals(jobType)) {
                job = (Job) SpringUtils.getBean(JobConstant.JOBNAME_HOS_HDW);
                jobExec(job,parameters,resultMap);
            }  else if (JobConstant.JOB_TYPE_BASE_INTER.equals(jobType)) {
                job = (Job) SpringUtils.getBean(JobConstant.JOBNAME_HOS_HDW_INTER);
                jobExec(job,parameters,resultMap);
            } else if (JobConstant.JOB_TYPE_DW.equals(jobType)) {
                job = (Job) SpringUtils.getBean(JobConstant.JOBNAME_ODS_DW);
                jobExec(job,parameters,resultMap);
            } else if (JobConstant.JOB_TYPE_DM.equals(jobType)) {
                Date dtBegin = parameters.getDate(JobConstant.JOB_DT_DAY_BEGIN);
                Date dtEnd =  parameters.getDate(JobConstant.JOB_DT_DAY_END);
                int counterStop = 0;//计算器，防止死循环，无实际意义
                job = (Job) SpringUtils.getBean(JobConstant.JOBNAME_DW_DM);
                if (job == null) return;
                JobParametersBuilder parametersBuilder = new JobParametersBuilder();
                parametersBuilder.addJobParameters(parameters);
                String dBegin;Date dtNext;
                while (dtBegin.before(dtEnd) && counterStop++ < 10000) {
                    dBegin = DateUtil.toDateStrByFormat(dtBegin, "yyyyMMdd");
                    parametersBuilder.addString(JobConstant.JOB_DAY,dBegin)
                            .addLong("time", System.currentTimeMillis())
                            .addString(JobConstant.JOB_DELSQL, parameters.getString(JobConstant.JOB_DELSQL).replaceAll("\\$\\{D_BEGIN\\}",dBegin))
                            .addString(JobConstant.JOB_QUERYSQL, parameters.getString(JobConstant.JOB_QUERYSQL).replaceAll("\\$\\{D_BEGIN\\}", dBegin));
                    jobExec(job,parametersBuilder.toJobParameters(),resultMap);
                    dtNext = DateUtil.getNextDate(dtBegin, 3);
                    dtBegin = dtNext;
                }
            }
        } catch (Exception e) {
            throw new BaseRuntimeException("任务运行异常，对应方法runJob",e);
        }
    }

    private void jobExec(Job job,JobParameters parameters,ConcurrentHashMap<Long, JobExecution> resultMap) throws BaseException {
        try {
            JobExecution run = jobLauncher.run(job, parameters);
            while(BatchStatus.STARTED.name().equals(run.getStatus().name()) || BatchStatus.STARTING.name().equals(run.getStatus().name())) {
                Thread.yield();
            }
            /*String tableName = parameters.getString(JobConstant.JOB_TABLENAME);
            if(tableName.toUpperCase().startsWith("DM_")) {
                String day = parameters.getString(JobConstant.JOB_DAY);

            }*/
            EtlTaskExecItem execItem;
            execItem = new EtlTaskExecItem();
            execItem.setIdExec(run.getJobParameters().getLong(JobConstant.JOB_ID_EXEC));
            execItem.setEuStatus(run.getStatus().name());
            execItem.setIdJob(run.getId());
            execItemService.insertEtlTaskExecItem(execItem);
            resultMap.put(run.getId(), run);
        } catch (Exception e) {
            throw new BaseException("job执行异常",e);
        }
    }
}
